package core;

import io.SocketChannelWrapper;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Timestamp;

import org.apache.log4j.Logger;

import protocol.Command;
import protocol.pv.PVCommand;
import util.DataUtil;
import dbop.DBManager;

/**
 * 
 * 
 * @author pdv
 *
 */
public class ProcessThread extends Thread {
	private static Logger logger = Logger.getLogger(ProcessThread.class);
	private static long fjcommandsendtime=0;
	private Connection dbconn;
	private Connection dbconn_noautocommit; //���Զ��ύ��DB 
	//add a autoreconnect db
	public Connection getAnalConn() {
	    return dbconn;
	}
	
	public Connection getAnalNoAutoCommitConn() {
	    return dbconn_noautocommit;
	}
	private Object lock = new Object();
	private volatile boolean isTerminated = false;
	private SocketChannelWrapper commClient = null;
	//0������ȼ���
	private CommandQueue queues = new CommandQueue();  //���ʵʱ��ݡ���ʷ���
	private CommandQueue fjqueue = new CommandQueue(); //fj����
	
	public ProcessThread(String name) {
		this.setName(name);
		queues.setMAXNUM(5000);
//		for (int i=0; i<queues.length; i++) {
//		    queues[i] = new CommandQueue();
//		}
		//connDB();
	}
	/**
     * 
     */
    private void connDB() {
        try {
            dbconn = DBManager.generateDBConn();
            dbconn_noautocommit = DBManager.generateDBConn();
            //logger.info("process thread "+getName()+" connect database success");
        } catch (SQLException e) {
            logger.error("connect database error--"+e.getMessage()+"--error code:"+e.getErrorCode());
        }
    }
    public void terminate() {
		isTerminated = true;
	}
	public synchronized void setCommClient(SocketChannelWrapper channel) {
		commClient = channel;
	}
	public synchronized SocketChannelWrapper getCommClient() {
		return commClient;
	}
	public void wakeup() {
		synchronized(lock) {
			lock.notify();
		}
	}
//	private Command getHighestPriorityCommand() {
//		for (int i=0; i<queues.length; i++) {
//	        Command cmd;
//			try {
//				cmd = queues[i].get();
//			} catch (Exception e) {
//				logger.error("get cmd error",e);
//			}
//			}
//	        if (cmd != null)
//	            return cmd;
//	    }
//	    return null;
//	}
	
	/**
	 * ִ�к���
	 */
	public void run() {
		logger.info("Process thread: "+getName()+" Started");
		while (!isTerminated) {
			Command command = null;
			try {
				command = queues.get();
			} catch (Exception e1) {
				// TODO Auto-generated catch block
				logger.error("take command error,e");
				command = null;
			}
			if (command != null) {
				//logger.info("Process thread: "+getName()+" Get frame -- "+command.toString());
				try {
					command.decode(this);  //
                    
                } catch (SQLException e) {
                	int errorCode = e.getErrorCode();
                    logger.error("Process thread: "+getName()+" Database operation error:"+e.getMessage()+", ErrorCode: "+errorCode, e);
                    if (DBManager.isDBDisconnectError(e)) {
                    	reconnDB();
                    }
                } catch (NullPointerException e) {
                	if (this.getAnalConn() == null)
                		reconnDB();
                	else
                		logger.error("unexpected NullPointerException "+e, e);
				} catch (Throwable e) {
                	logger.error("Process thread: "+getName()+" runtime error", e);
                }
			}
		}
		try {
			if (this.getAnalConn() != null)
				this.getAnalConn().close();
		} catch (SQLException e) {
			e.printStackTrace();
		}
		logger.info("process thread "+this.getName()+" ended");
	}

	/**
     * ������ݿ�
     */
    public void reconnDB() {
        logger.info("Process thread: "+getName()+". Start reconn db");
        try {
        	if (dbconn != null)
        		dbconn.close();
        	if (dbconn_noautocommit != null)
        		dbconn_noautocommit.close();
        } catch (SQLException e) {
        }
        connDB();        
    }
    /**
     *������� 
     */
	public void feedCommand(Command command) {
		this.queues.add(command);
		
	}


	public boolean sendToComm(String s) {
		return direct_sendToComm(s);
	}
	public long sendBytes(byte[] buff){
		logger.info("write to host "+ DataUtil.BytesToHexStr(buff));
		ByteBuffer srcs= ByteBuffer.wrap(buff);
		int  res=-1;
		try {
			SocketChannel socketChannel = this.getCommClient().getChannel();
			if(socketChannel.isConnected())
			{
				res = this.getCommClient().getChannel().write(srcs);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		return res;
		
	}
	public long sendIframe(byte[] buff){
		this.sendIndex++;
		if (sendIndex>=32767){
			sendIndex=0;
		}
		return this.sendBytes(buff);
	}
	
	public long sendUframe(byte[] buff){
		this.sendIndex++;
		if (sendIndex>=32767){
			sendIndex=0;
		}
		return this.sendBytes(buff);
	}
	
	public boolean sendToCommDirect(String s) {
		return direct_sendToComm(s);
	}


	private boolean direct_sendToComm(String s) {
		boolean res;
		SocketChannelWrapper wrap = this.getCommClient();
		if (wrap == null) {
			res = false;
		} else
			res = wrap.write(s);
		if (!res)
			logger.error("write to comm "+this.getName()+" "+s+" failed");
		return res;
	}
	private int hostRecvIndex;
	public int getHostRecvIndex() {
		return hostRecvIndex;
	}

	public void setHostRecvIndex(int hostRecvIndex) {
		this.hostRecvIndex = hostRecvIndex;
	}
	private int rcvIndex=0;
	private int sendIndex=0;
	public int getRcvIndex() {
		return rcvIndex;
	}

	public void setRcvIndex(int rcvIndex) {
		this.rcvIndex = rcvIndex;
	}

	public int getSendIndex() {
		return sendIndex;
	}

	public void setSendIndex(int sendIndex) {
		this.sendIndex = sendIndex;
	}
	
	
}
